package io.openmessaging.io;

import io.openmessaging.utils.Assert;
import io.openmessaging.utils.AtomicLong;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.FileSystems;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;

/**
 * 提高 size() 方法速度，
 * 代价是只能追加写，不能随机写
 */
public class FileChannelWrapper implements IO {
    public static final class FileChannelException extends RuntimeException {
        public FileChannelException(Exception e) {
            super(e);
        }
    }

    private final Path path;
    private final FileChannel channel;
    private AtomicLong writePos = new AtomicLong();

    public FileChannelWrapper(String path) {
        try {
            this.path = FileSystems.getDefault().getPath(path);
            this.channel = FileChannel.open(this.path, StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE);
            writePos.set(channel.size());
        } catch (IOException e) {
            throw new FileChannelException(e);
        }

    }

    @Override
    public long size() {
        return writePos.get();
    }

    @Override
    public void delete() {
        try {
            channel.close();
            File file = path.toFile();
            if (file.isFile())
                Assert.isTrue(file.delete());
        } catch (IOException e) {
            throw new FileChannelException(e);
        }
    }

    /**
     * 读满并flip dst缓冲区
     * 如果剩余数据不足以装满缓冲区，会抛出异常
     */
    @Override
    public int readAndFlip(ByteBuffer dst, long position) {
        try {
            return read(dst, position);
        } finally {
            dst.flip();
        }
    }

    @Override
    public int read(ByteBuffer dst, long position) {
        try {
            int i = 0;
            while (dst.hasRemaining())
                i += channel.read(dst, position + i);
            return i;
        } catch (IOException e) {
            throw new FileChannelException(e);
        }
    }


    @Override
    public long write(ByteBuffer... buffers) {
        try {
            long size = 0;
            for (ByteBuffer buffer : buffers) {
                size += buffer.remaining();
            }
            long position = writePos.getAndAdd(size);
            long i = position;
            for (ByteBuffer buffer : buffers) {
                while (buffer.hasRemaining())
                    i += channel.write(buffer, i);
//                Thread.sleep(0);
            }
            return position;
        } catch (Exception e) {
            throw new FileChannelException(e);
        }
    }

    @Override
    public void force() {
        try {
            channel.force(false);
//            Thread.sleep(0);
        } catch (Exception e) {
            throw new FileChannelException(e);
        }
    }

    @Override
    public void close() {
        try {
            channel.close();
        } catch (IOException e) {
            throw new FileChannelException(e);
        }
    }
}
